package com.open.source.kafka.interceptor;

import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.Map;
import java.util.UUID;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;

/**
 * interceptor 可能运行在多个线程中，因此在具体实现时用户需要自行确保线程安 全。另外，若指定了多个 interceptor ，则 producer 将按照指定顺序调用它们，同时把每个
 * interceptor 中捕获的异常记录到错误日志中而不是向上传递。这在使用过程中要特别留意。
 *
 * @author ZonLen since on 2022/7/29 下午3:26
 */
public class KafkaProducerTraceInterceptor implements ProducerInterceptor<String, String> {

  /**
   * <p>
   * 该方法封装进 KafkaProducer.send 法中，即它运行在用户 线程中 producer 确保在消息被序列化以计算分区前调用该方法 用户可以在该方
   * 法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区，否则会影响目 标分区的计算
   * <p>
   */
  @Override
  public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    final ArrayList<Header> headers = new ArrayList<>();
    headers.add(new RecordHeader("host-ip", KafkaProducerTraceInterceptor.getHostIp().getBytes()));
    headers.add(new RecordHeader("trance-id", UUID.randomUUID().toString().getBytes()));
    headers.add(new RecordHeader("sender-model", "opensource-kafka-producter".getBytes()));
    return new ProducerRecord<>(record.topic(),
        record.partition(), record.key(), record.value(), headers);
  }

  /**
   * <p>
   * 该方法会在消息被应答之前或消息发 送失败时调用，井且通常都是在 producer 回调逻辑触发之前 onAcknowledgement 行在 producer 1/0
   * 线程中，因此不要在该方法中放入很“重”的逻辑，否则会拖慢 producer 的消息发送效率。
   * <p>
   */
  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

  }

  @Override
  public void close() {

  }

  @Override
  public void configure(Map<String, ?> configs) {

  }


  private static String getHostIp() {
    try {
      Enumeration<NetworkInterface> allNetInterfaces = NetworkInterface.getNetworkInterfaces();
      while (allNetInterfaces.hasMoreElements()) {
        NetworkInterface netInterface = allNetInterfaces.nextElement();
        Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
        while (addresses.hasMoreElements()) {
          InetAddress ip = addresses.nextElement();
          if (ip == null) {
            return "";
          }
          if (ip instanceof Inet4Address
              //loopback地址即本机地址，IPv4的loopback范围是127.0.0.0 ~ 127.255.255.255
              && !ip.isLoopbackAddress()
              && !ip.getHostAddress().contains(":")) {
            return ip.getHostAddress();
          }
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
    return "";
  }

}
